package mapReduce.demo07_join.demo02_productOrderMapJoin;

import mapReduce.util.JobSubmit;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

public class ProductOrderMapJoinApp {

    public static class POMMapper extends Mapper<LongWritable, Text,Text, NullWritable>{

        Text outKey = new Text();
        Map<String,String> productInfo = new HashMap<>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
            //从上下文中获取所有的 cacheFile
            URI[] cacheFiles = context.getCacheFiles();
            //在此逻辑中，只需要一个 cacheFile (商品信息文件)
            URI productFile = cacheFiles[0];
            //获取此文件的绝对路径
            String productFilePath = productFile.getPath();
            //创建了一个商品信息文件的 文件字符输入流
            FileReader fr = new FileReader(productFilePath);
            //把节点流封装成增强流
            BufferedReader br = new BufferedReader(fr);

            //读取商品信息数据并把数据封装成Map
            String line = "";
            while((line = br.readLine())!=null){
                String[] fields = line.split("\t");
                productInfo.put(fields[0],fields[1]);
            }

        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            String lineKey = fields[0]+"\t"+fields[1]+"\t"+ productInfo.get(fields[1]) +"\t"+fields[2];
            outKey.set(lineKey);

            context.write(outKey,NullWritable.get());

        }
    }

    public static void main(String[] args) throws URISyntaxException {
        JobSubmit.submitBaseJob(ProductOrderMapJoinApp.class,args,0,new URI[]{new URI("file:///D:/data/data_hadoop/join/input_product/product.txt")});
    }

}
